package com.mqtt.handler;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.mqtt.model.MqttMessage;
import com.mqtt.model.DeviceInfo;
import com.mqtt.service.impl.KafkaProducerService;
import com.mqtt.service.impl.MessageProcessService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import com.mqtt.service.impl.DeviceSessionService;
import com.mqtt.handler.DeviceMessageHandler;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
 * MQTT消息处理器
 */
@Slf4j
@Component
public class MqttMessageHandler {
    
    @Autowired
    private DeviceMessageHandler deviceMessageHandler;
    
    @Autowired
    private MessageProcessService messageProcessService;
    
    @Autowired
    private DeviceSessionService deviceSessionService;
    
    @Autowired
    private KafkaProducerService kafkaProducerService;
    
    /**
     * 处理MQTT接收到的消息
     */
    @ServiceActivator(inputChannel = "mqttInputChannel")
    @Async("mqttMessageExecutor")
    public void handleMessage(Message<?> message) {
        try {
            MessageHeaders headers = message.getHeaders();
            String topic = String.valueOf(headers.get("mqtt_receivedTopic"));
            String payload = String.valueOf(message.getPayload());
            Integer qos = (Integer) headers.get("mqtt_receivedQos");
            
            log.info("Received MQTT message - Topic: {}, QoS: {}, Payload: {}", topic, qos, payload);
            
            // 创建MQTT消息对象
            MqttMessage mqttMessage = new MqttMessage();
            mqttMessage.setTopic(topic);
            mqttMessage.setPayload(payload);
            mqttMessage.setQos(qos);
            mqttMessage.setTimestamp(new Date());
            
            // 解析topic获取设备信息
            DeviceInfo deviceInfo = parseDeviceFromTopic(topic);
            if (deviceInfo != null) {
                mqttMessage.setDeviceId(deviceInfo.getDeviceId());
                mqttMessage.setDeviceType(deviceInfo.getDeviceType());
                
                // 更新设备会话
                deviceSessionService.updateSession(deviceInfo.getDeviceId(), topic);
            }
            
            // 根据topic路由处理
            routeMessage(mqttMessage);
            
        } catch (Exception e) {
            log.error("Error handling MQTT message", e);
        }
    }
    
    /**
     * 消息路由处理
     */
    private void routeMessage(MqttMessage mqttMessage) {
        String topic = mqttMessage.getTopic();
        
        try {
            if (topic.contains("/data")) {
                // 设备数据上报
                handleDeviceData(mqttMessage);
            } else if (topic.contains("/status")) {
                // 设备状态上报
                handleDeviceStatus(mqttMessage);
            } else if (topic.contains("/command")) {
                // 命令响应
                handleCommandResponse(mqttMessage);
            } else if (topic.contains("/alarm")) {
                // 告警消息
                handleAlarmMessage(mqttMessage);
            } else if (topic.contains("/heartbeat")) {
                // 心跳消息
                handleHeartbeat(mqttMessage);
            } else if (topic.contains("/register")) {
                // 设备注册
                handleDeviceRegister(mqttMessage);
            } else {
                // 其他消息
                handleOtherMessage(mqttMessage);
            }
        } catch (Exception e) {
            log.error("Error routing message for topic: {}", topic, e);
        }
    }
    
    /**
     * 处理设备数据
     */
    private void handleDeviceData(MqttMessage mqttMessage) {
        log.debug("Processing device data: {}", mqttMessage.getDeviceId());
        
        // 处理消息
        Map<String, Object> processedData = messageProcessService.processDeviceData(mqttMessage);
        
        // 发送到Kafka
        kafkaProducerService.sendToKafka("device-data", mqttMessage.getDeviceId(), processedData);
        
        // 调用设备消息处理器
        deviceMessageHandler.handleDataMessage(mqttMessage);
    }
    
    /**
     * 处理设备状态
     */
    private void handleDeviceStatus(MqttMessage mqttMessage) {
        log.debug("Processing device status: {}", mqttMessage.getDeviceId());
        
        // 解析状态信息
        JSONObject statusJson = JSON.parseObject(mqttMessage.getPayload());
        String status = statusJson.getString("status");
        
        // 更新设备状态
        deviceSessionService.updateDeviceStatus(mqttMessage.getDeviceId(), status);
        
        // 发送到Kafka
        kafkaProducerService.sendToKafka("device-status", mqttMessage.getDeviceId(), statusJson);
        
        // 调用设备消息处理器
        deviceMessageHandler.handleStatusMessage(mqttMessage);
    }
    
    /**
     * 处理命令响应
     */
    private void handleCommandResponse(MqttMessage mqttMessage) {
        log.debug("Processing command response: {}", mqttMessage.getDeviceId());
        
        // 处理命令响应
        deviceMessageHandler.handleCommandResponse(mqttMessage);
        
        // 发送到Kafka
        Map<String, Object> response = new HashMap<>();
        response.put("deviceId", mqttMessage.getDeviceId());
        response.put("response", mqttMessage.getPayload());
        response.put("timestamp", mqttMessage.getTimestamp());
        
        kafkaProducerService.sendToKafka("command-response", mqttMessage.getDeviceId(), response);
    }
    
    /**
     * 处理告警消息
     */
    private void handleAlarmMessage(MqttMessage mqttMessage) {
        log.warn("Alarm message from device: {}", mqttMessage.getDeviceId());
        
        // 解析告警信息
        JSONObject alarmJson = JSON.parseObject(mqttMessage.getPayload());
        
        // 发送到Kafka告警topic
        kafkaProducerService.sendToKafka("device-alarm", mqttMessage.getDeviceId(), alarmJson);
        
        // 调用设备消息处理器
        deviceMessageHandler.handleAlarmMessage(mqttMessage);
    }
    
    /**
     * 处理心跳消息
     */
    private void handleHeartbeat(MqttMessage mqttMessage) {
        log.debug("Heartbeat from device: {}", mqttMessage.getDeviceId());
        
        // 更新设备最后活跃时间
        deviceSessionService.updateHeartbeat(mqttMessage.getDeviceId());
        
        // 可选：发送到Kafka
        if (shouldForwardHeartbeat()) {
            Map<String, Object> heartbeat = new HashMap<>();
            heartbeat.put("deviceId", mqttMessage.getDeviceId());
            heartbeat.put("timestamp", mqttMessage.getTimestamp());
            
            kafkaProducerService.sendToKafka("device-heartbeat", mqttMessage.getDeviceId(), heartbeat);
        }
    }
    
    /**
     * 处理设备注册
     */
    private void handleDeviceRegister(MqttMessage mqttMessage) {
        log.info("Device registration: {}", mqttMessage.getDeviceId());
        
        // 解析注册信息
        JSONObject registerInfo = JSON.parseObject(mqttMessage.getPayload());
        
        // 创建设备会话
        deviceSessionService.createSession(mqttMessage.getDeviceId(), registerInfo);
        
        // 发送到Kafka
        kafkaProducerService.sendToKafka("device-register", mqttMessage.getDeviceId(), registerInfo);
        
        // 调用设备消息处理器
        deviceMessageHandler.handleRegisterMessage(mqttMessage);
    }
    
    /**
     * 处理其他消息
     */
    private void handleOtherMessage(MqttMessage mqttMessage) {
        log.info("Other message from topic: {}", mqttMessage.getTopic());
        
        // 发送到Kafka默认topic
        Map<String, Object> message = new HashMap<>();
        message.put("topic", mqttMessage.getTopic());
        message.put("deviceId", mqttMessage.getDeviceId());
        message.put("payload", mqttMessage.getPayload());
        message.put("timestamp", mqttMessage.getTimestamp());
        
        kafkaProducerService.sendToKafka("mqtt-other", mqttMessage.getDeviceId(), message);
    }
    
    /**
     * 从topic中解析设备信息
     */
    private DeviceInfo parseDeviceFromTopic(String topic) {
        try {
            // 假设topic格式: device/{deviceId}/data
            String[] parts = topic.split("/");
            if (parts.length >= 2 && "device".equals(parts[0])) {
                DeviceInfo deviceInfo = new DeviceInfo();
                deviceInfo.setDeviceId(parts[1]);
                
                // 可以从Redis或数据库获取更多设备信息
                return deviceInfo;
            }
        } catch (Exception e) {
            log.error("Error parsing device from topic: {}", topic, e);
        }
        return null;
    }
    
    /**
     * 是否转发心跳消息到Kafka
     */
    private boolean shouldForwardHeartbeat() {
        // 可以根据配置或业务逻辑决定
        return false;
    }
}